package com.yjy.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.List;

@Component
public class MessageConsumer implements MessageListenerConcurrently {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("v4-consumer-group");


    @PostConstruct
    public void start() {
        try {
            consumer.setNamesrvAddr(namesrvAddr);
            //从消息队列头部开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
            //设置集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //订阅主题
            consumer.subscribe("test-topic-local06", "*");

            //注册消息监听器
            consumer.registerMessageListener(this);

            //启动消费端
            consumer.start();

            logger.info("Message Consumer Start...");
            System.err.println("Message Consumer Start...");
        } catch (MQClientException e) {
            logger.error("Message Consumer Start Error!!", e);
        }

    }

    // messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    // 失败重试频率：1s 5s 5s 5s 10s 10s 10s 30s 30s 30s 1m 1m 1m 1m 2m 2m 2m 一共通知15次，超过后自动返回成功
    public static final Integer[] retryTime = {1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 4, 5, 5, 5, 5};

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        MessageExt message = msgs.get(0);
        try {
            //逐条消费
            String messageBody = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
            logger.info("消息：{} 时间：{} 次数：{}", messageBody, LocalDateTime.now(), message.getReconsumeTimes());

            //模拟业务异常
            int i = 1 / 0;
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            int reconsumeTimes = message.getReconsumeTimes();
            if (reconsumeTimes >= 15) {
                System.out.println("重试15次后返回成功");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            context.setDelayLevelWhenNextConsume(retryTime[reconsumeTimes]);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}